package com.liang.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class TimeInterceptor implements ProducerInterceptor<String,String> {

    int success;
    int error;

    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {

        return new ProducerRecord<String, String>(producerRecord.topic(),producerRecord.partition(),
                producerRecord.timestamp(),producerRecord.key(),System.currentTimeMillis()+"-"+producerRecord.value());
    }

    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (e==null){
            success++;
        }else {
            error++;
        }
    }

    public void close() {
        System.out.println("success = " + success);
        System.out.println("error = " + error);
    }

    public void configure(Map<String, ?> map) {

    }
}
